/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *    Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.transfer;

import com.dtdream.emr.transmission.io.Request;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.util.List;

public class Decoder extends ByteToMessageDecoder {

    private static final Logger LOG = LoggerFactory.getLogger(Decoder.class);
    //Fixed length for a message length value;
    private static final int HEAD_LENGTH = 4;

    private Kryo kryo = new Kryo();

    private Request converToMessage(byte[] in) {
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(in);
        Input input = new Input(byteArrayInputStream);

        return kryo.readObject(input, Request.class);
    }

    /**
     * @param channelHandlerContext
     * @param byteBuf
     * @param list
     * @throws Exception
     */
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        int readableBytes = byteBuf.readableBytes();

        if (readableBytes >= HEAD_LENGTH) {
            byteBuf.markReaderIndex();

            int dataLength = byteBuf.readInt();
            if (dataLength < 0) {
                /* Data error */
                channelHandlerContext.close();
            }

            /* update readable bytes */
            readableBytes = byteBuf.readableBytes();
            LOG.debug("Readable bytes " + readableBytes + " dataLength: " + dataLength);

            if (dataLength <= readableBytes) {
                byte[] body = new byte[dataLength];
                byteBuf.readBytes(body);
                list.add(converToMessage(body));
            } else {
                byteBuf.resetReaderIndex();
            }
        }
    }
}
